package cn.getech.data.development.utils

import cn.getech.data.development.bean.dag.FlinkStreamTypeObj
import org.slf4j.{Logger, LoggerFactory}

/**
 * dag node转换成sql
 */
object FlinkSQLAnalysisUtils {

  final object Common {
    final val space = " "
    final val bracketLeft = "("
    final val bracketRight = ")"
    final val equal = "="
    final val from = "FROM"
    final val select = "SELECT"
    final val asterisk = "*"
    final val comma = ","
    final val insert = "INSERT INTO"
    final val createView = "CREATE VIEW"
    final val as = "AS"
    final val where = "WHERE"
    final val row_number = "ROW_NUMBER()"
    final val over = "OVER"
    final val orderBy = "ORDER BY"
    final val partitionBy = "PARTITION BY"
    final val groupBy = "GROUP BY"
    final val interval = "INTERVAL"
    final val singleQuotes = "'"
  }

  private val logger: Logger = LoggerFactory.getLogger(this.getClass)

  /**
   * select sql
   *
   * @param obj json
   * @return (resultText, insertName)
   */
  def selectSQL(obj: FlinkStreamTypeObj, filterBlock: () => String = null, distinctBlock: () => String = null, groupByBlock: () => String = null): String = {

    val sb = new StringBuilder
    // 头部: select...
    sb.append(Common.select).append(Common.space)
    val tableName: String = obj.tableName

    //TUMBLE_START(rowtime, INTERVAL '1' DAY) as wStart
    obj.fields.foreach(x => {
      val function = x.function
      val alias = x.alias

      sb.append(
        stringOrElse(!function.isEmpty,
          function,
          x.field_name)
      )
      sb.append(
        stringOrElse(
          !alias.isEmpty,
          Common.space + Common.as + Common.space + alias + Common.comma,
          Common.comma)
      )
    })

    // 替换最后一个符号
    sb.replace(sb.length - 1, sb.length, Common.space)

    // 尾部: from...
    sb.append(Common.from).append(Common.space)

    // distinct直接拼接整段数据
    if (distinctBlock != null)
      sb.append(distinctBlock())
    else {
      sb.append(tableName).append(Common.space)
      // filter
      if (filterBlock != null)
        sb.append(filterBlock())
      // group by
      if (groupByBlock != null)
        sb.append(groupByBlock())
    }

    // TODO insert 暂时不添加。。。
    //    var insertName = obj.getString("insertName")
    //    if (insertName.isEmpty)
    //      insertName = tableName
    //    (sb.toString(), insertName)
    logger.info("sql: {}" + sb.toString())
    sb.toString()
  }

  /**
   * filter sql
   *
   * @param obj json
   * @return WHERE expression
   */
  def filterSQL(obj: FlinkStreamTypeObj): String = {
    selectSQL(obj, { () =>
      val expression = obj.fieldText
      stringOrElse(!expression.isEmpty, Common.space + Common.where + Common.space + expression, "")
    })
  }

  /**
   * group by
   *
   * @param obj
   * @return GROUP BY TUMBLE(windowName, INTERVAL 'interval' unit), groupName
   */
  def groupBySQL(obj: FlinkStreamTypeObj): String = {
    selectSQL(obj, null, null, { () =>
      var res = Common.groupBy + Common.space
      val isWindow = obj.isWindow
      val groupName = obj.groupName

      if (isWindow) {
        val windowName = obj.windowName
        val windowType = obj.windowType
        val interval = obj.interval
        val unit = obj.unit
        res += windowType + Common.bracketLeft +
          windowName + Common.comma +
          Common.interval + Common.space +
          Common.singleQuotes + interval + Common.singleQuotes + Common.space +
          unit + Common.bracketRight
        if (!groupName.isEmpty)
          res += Common.comma + Common.space + groupName
      } else {
        res += groupName
      }
      res
    })
  }

  /**
   * distinct sql
   *
   * @param obj json
   * @return (SELECT *, funcName OVER(PARTITION BY parameter1 ORDER BY parameter2 sort) AS alias FROM tableName) WHERE alias = 1
   */
  def distinctSQL(obj: FlinkStreamTypeObj): String = {
    def paraExpr(): String = {
      Common.groupBy + Common.space + obj.fields.map(_.field_name).mkString(", ")
    }

    selectSQL(obj, null, { () =>
      val alias: String = obj.alias_partition
      val sql: String = topNOver(
        Common.row_number,
        obj.partition,
        obj.order,
        alias,
        obj.tableName)
      Common.bracketLeft + sql + Common.bracketRight + Common.space +
        Common.where + Common.space + alias + Common.space + Common.equal + Common.space + "1" + Common.space + paraExpr()
    })
  }

  /**
   * @param funcName  row_number() | rank() | dense_rank()
   * @param partition parameter1
   * @param order     parameter2
   * @param alias     别名
   * @param tableName 表名
   * @return SELECT *, funcName OVER(PARTITION BY parameter1 ORDER BY parameter2 desc) AS alias FROM tableName GROUP BY a,b,c,d
   */
  private def topNOver(funcName: String, partition: String, order: String, alias: String = "", tableName: String): String = {
    var res: String = Common.select + Common.space + Common.asterisk + Common.comma + Common.space +
      partitionOver(funcName, partition, order)

    if (!alias.isEmpty) res += Common.space + Common.as + Common.space + alias + Common.space
    res += Common.from + Common.space + tableName
    res
  }

  /**
   * create view
   *
   * @param tableName table name
   * @param sql       sql
   * @return create view tableName as select * from xxx
   */
  def createViewSQL(tableName: String, sql: String): String = {
    var res = ""
    if (!tableName.isEmpty)
      res = s"${Common.createView} ${tableName} ${Common.as} "
    res ++ sql
  }

  /**
   * insert table
   *
   * @param insertName insert table name
   * @param sql        sql
   * @return insert into tableName select xxx from xxx
   */
  def insertSQL(insertName: String, sql: String): String = {
    if (sql.isEmpty) ""
    stringOrElse(!insertName.isEmpty, s"${Common.insert} ${insertName} ${sql}", "")
  }

  def select(tableName: String): String = {
    "select * from " + tableName
  }

  /**
   * @param funcName  row_number() | rank() | dense_rank()
   * @param partition parameter1
   * @param order     parameter2
   * @return ROW_NUMBER() OVER(PARTITION BY parameter1 ORDER BY parameter2 desc)
   */
  private def partitionOver(funcName: String, partition: String, order: String): String = {
    s"$funcName ${Common.over}(${Common.partitionBy} ${partition} ${Common.orderBy} $order )"
  }

  private def stringOrElse(flag: Boolean, string: String, other: String): String = {
    if (flag)
      string
    else
      other
  }

}
